<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>

    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <meta content="Cask Data, Inc." name="author" />
<meta content="Copyright © 2016-2017 Cask Data, Inc." name="copyright" />


    <meta name="git_release" content="6.1.1">
    <meta name="git_hash" content="05fbac36f9f7aadeb44f5728cea35136dbc243e5">
    <meta name="git_timestamp" content="2020-02-09 08:22:47 +0800">
    <title>Creating a Plugin</title>

    <link rel="stylesheet" href="../../_static/cdap-bootstrap.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/bootstrap-3.3.6/css/bootstrap.min.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/bootstrap-3.3.6/css/bootstrap-theme.min.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/css/bootstrap-sphinx.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/css/cdap-dynamicscrollspy-4.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/css/jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/css/cdap-jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/css/abixTreeList-2.css" type="text/css" />
    <link rel="stylesheet" href="../../_static/cdap-bootstrap.css" type="text/css" />

    <script type="text/javascript">
      var DOCUMENTATION_OPTIONS = {
        URL_ROOT:    '',
        VERSION:     '6.1.1',
        COLLAPSE_INDEX: false,
        FILE_SUFFIX: '.html',
        HAS_SOURCE:  false
      };
    </script>
    <script type="text/javascript" src="../../_static/jquery.js"></script>
    <script type="text/javascript" src="../../_static/underscore.js"></script>
    <script type="text/javascript" src="../../_static/doctools.js"></script>
    <script type="text/javascript" src="../../_static/language_data.js"></script>

    <link rel="shortcut icon" href="../../_static/favicon.ico"/>
    <link rel="index" title="Index" href="../../genindex.html" />
    <link rel="search" title="Search" href="../../search.html" />
    <link rel="top" title="Cask Data Application Platform 6.1.1 Documentation" href="../../index.html" />
    <link rel="up" title="Developing Plugins" href="index.html" />
    <link rel="next" title="Plugin Presentation" href="presentation-plugins.html" />
    <link rel="prev" title="Plugin Basics" href="plugin-basics.html" />
    <!-- block extrahead -->
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge,chrome=1'>
    <meta name='viewport' content='width=device-width, initial-scale=1.0, maximum-scale=1'>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <!-- block extrahead end -->

</head>
<body role="document">

<!-- block navbar -->
<div id="navbar" class="navbar navbar-inverse navbar-default navbar-fixed-top">
    <div class="container-fluid">
      <div class="row">
        <div class="navbar-header">
          <!-- .btn-navbar is used as the toggle for collapsed navbar content -->
          <a class="navbar-brand" href="../../table-of-contents/../../index.html">
            <span><img alt="CDAP logo" src="../../_static/cdap_logo.svg"/></span>
          </a>

          <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".nav-collapse">
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
          </button>

          <div class="pull-right">
            <div class="dropdown version-dropdown">
              <a href="#" class="dropdown-toggle" data-toggle="dropdown"
                role="button" aria-haspopup="true" aria-expanded="false">
                v 6.1.1 <span class="caret"></span>
              </a>
              <ul class="dropdown-menu">
                <li><a href="//docs.cdap.io/cdap/5.1.2/en/index.html">v 5.1.2</a></li>
                <li><a href="//docs.cdap.io/cdap/4.3.4/en/index.html">v 4.3.4</a></li>
              </ul>
            </div>
          </div>
          <form class="navbar-form navbar-right navbar-search" action="../../search.html" method="get">
            <div class="form-group">
              <div class="navbar-search-image material-icons"></div>
              <input type="text" name="q" class="form-control" placeholder="  Search" />
            </div>
            <input type="hidden" name="check_keywords" value="yes" />
            <input type="hidden" name="area" value="default" />
          </form>

          <div class="collapse navbar-collapse nav-collapse navbar-right navbar-navigation">
            <ul class="nav navbar-nav"><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../../table-of-contents/../../index.html">简介</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link current" href="../../table-of-contents/../../guides.html">手册</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../../table-of-contents/../../reference-manual/index.html">参考</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../../table-of-contents/../../faqs/index.html">帮助</a></li>
            </ul>
          </div>

        </div>
      </div>
    </div>
  </div><!-- block navbar end -->
<!-- block main content -->
<div class="main-container container">
  <div class="row"><div class="col-md-2">
      <div id="sidebar" class="bs-sidenav scrollable-y-outside" role="complementary">
<!-- theme_manual: developer-manual -->
<!-- theme_manual_highlight: guides -->
<!-- sidebar_title_link: ../../table-of-contents/../../guides.html -->

  <div role="note" aria-label="manuals links"><h3><a href="../../table-of-contents/../../guides.html">Guides</a></h3>

    <ul class="this-page-menu">
      <li class="toctree-l1"><a href="../../table-of-contents/../../user-guide/index.html" rel="nofollow">用户手册</a>
      </li>
      <li class="toctree-l1"><b><a href="../../table-of-contents/../../developer-manual/index.html" rel="nofollow">开发手册</a></b>
      <nav class="pagenav">
      <ul class="current">
<li class="toctree-l1"><a class="reference internal" href="../../index.html"> 简介</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../getting-started/index.html"> 入门指南</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../getting-started/sandbox/index.html">CDAP Sandbox</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../../getting-started/sandbox/zip.html">二进制 Zip 文件</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../getting-started/sandbox/zip.html#cdap-sandbox">启动和停止 CDAP Sandbox</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../getting-started/sandbox/virtual-machine.html">虚拟机镜像</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../getting-started/sandbox/docker.html">Docker 镜像</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../../getting-started/quick-start.html">快速入门</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../getting-started/dev-env.html">搭建开发环境</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../getting-started/start-stop-cdap.html">启动和停止 CDAP</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../getting-started/building-apps.html">构建并运行应用</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../overview/index.html"> 概述</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../overview/anatomy.html"> 大数据应用剖析</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../overview/modes.html"> 模式和组件</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../overview/abstractions.html"> 核心概念</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../overview/interfaces.html"> 编程接口</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../building-blocks/index.html"> 抽象概念</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/core.html"> Core Abstractions</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/applications.html"> Applications</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/datasets/index.html"> Datasets</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/overview.html"> Overview</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/table.html"> Table API</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/fileset.html"> FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/partitioned-fileset.html"> Partitioned FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/time-partitioned-fileset.html"> TimePartitioned FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/system-custom.html"> System and Custom Datasets</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/permissions.html"> Dataset Permissions</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../building-blocks/datasets/cube.html"> Cube Dataset</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/mapreduce-programs.html"> MapReduce Programs</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/plugins.html"> Plugins</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/schedules.html"> Schedules</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/secure-keys.html"> Secure Keys</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/services.html"> Services</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/spark-programs.html"> Spark Programs</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/workers.html"> Workers</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/workflows.html"> Workflows</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/artifacts.html"> Artifacts</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/program-lifecycle.html"> Program Lifecycle</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/namespaces.html"> Namespaces</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/transaction-system.html"> Transaction System</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../building-blocks/transactional-messaging-system.html"> Transactional Messaging System</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../metadata/index.html"> 元数据</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/system-metadata.html"> System Metadata</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/discovery-lineage.html"> Discovery and Lineage</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/field-lineage.html"> Field Level Lineage</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/audit-logging.html"> Audit Logging</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/metadata-ui.html"> CDAP Metadata UI</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../metadata/programmatic-metadata.html"> Accessing metadata programmatically</a></li>
</ul>
</li>
<li class="toctree-l1 current"><a class="reference internal" href="../index.html"> 数据流管道</a><ul class="current">
<li class="toctree-l2"><a class="reference internal" href="../concepts-design.html"> Concepts and Design</a></li>
<li class="toctree-l2"><a class="reference internal" href="../getting-started.html"> Getting Started</a></li>
<li class="toctree-l2"><a class="reference internal" href="../studio.html"> CDAP Studio</a></li>
<li class="toctree-l2"><a class="reference internal" href="../creating-pipelines.html"> Creating Pipelines</a></li>
<li class="toctree-l2"><a class="reference internal" href="../running-pipelines.html"> Running Pipelines</a></li>
<li class="toctree-l2"><a class="reference internal" href="../plugin-management.html"> Plugin Management</a></li>
<li class="toctree-l2"><a class="reference internal" href="../plugins/index.html"> Plugin Reference</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../plugins/actions/index.html"> Action Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/sources/index.html"> Source Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/transforms/index.html"> Transform Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/analytics/index.html"> Analytic Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/sinks/index.html"> Sink Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/shared-plugins/index.html"> Shared Plugins</a><ul>
<li class="toctree-l4"><a class="reference internal" href="../plugins/shared-plugins/core.html">CoreValidator</a></li>
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../plugins/post-run-plugins/index.html"> Post-run Plugins</a><ul class="simple">
</ul>
</li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../developing-pipelines.html"> Developing Pipelines</a></li>
<li class="toctree-l2 current"><a class="reference internal" href="index.html"> Developing Plugins</a><ul class="current">
<li class="toctree-l3"><a class="reference internal" href="plugin-basics.html">Plugin Basics</a></li>
<li class="toctree-l3 current"><a class="current reference internal" href="#">Creating a Plugin</a></li>
<li class="toctree-l3"><a class="reference internal" href="presentation-plugins.html">Plugin Presentation</a></li>
<li class="toctree-l3"><a class="reference internal" href="testing-plugins.html">Testing Plugins</a></li>
<li class="toctree-l3"><a class="reference internal" href="packaging-plugins.html">Packaging Plugins</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../how-cdap-pipelines-work.html"> How CDAP Pipelines Work</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../cloud-runtimes/index.html"> 云平台运行</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../cloud-runtimes/concepts/index.html"> Concepts</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../cloud-runtimes/provisioners/index.html"> Provisioners</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/provisioners/gcp-dataproc.html">Google Dataproc</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/provisioners/aws-emr.html">Amazon Elastic MapReduce</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/provisioners/remote-hadoop.html">Remote Hadoop</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../../cloud-runtimes/profiles/index.html"> Profiles</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/profiles/creating-profiles.html">Creating Profiles</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/profiles/assigning-profiles.html">Assigning Profiles</a></li>
<li class="toctree-l3"><a class="reference internal" href="../../cloud-runtimes/profiles/admin-controls.html">Admin Controls</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../../cloud-runtimes/example/index.html"> Example</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../security/index.html"> 安全</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../security/client-authentication.html">Client Authentication</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../security/cdap-authentication-clients-java.html">CDAP Authentication Client for Java</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../security/cdap-authentication-clients-python.html">CDAP Authentication Client for Python</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../security/custom-authentication.html">Custom Authentication</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../security/authorization-extensions.html">Authorization Extensions</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../testing/index.html"> 测试和调试</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../testing/testing.html"> Testing a CDAP Application</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../testing/debugging.html"> Debugging</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../testing/troubleshooting.html"> Troubleshooting</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../ingesting-tools/index.html"> 数据融合</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../ingesting-tools/cdap-stream-clients-java.html">CDAP Stream Client for Java</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../ingesting-tools/cdap-stream-clients-python.html">CDAP Stream Client for Python</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../ingesting-tools/cdap-stream-clients-ruby.html">CDAP Stream Client for Ruby</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../ingesting-tools/cdap-flume.html">CDAP Flume</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../data-exploration/index.html"> 数据探索</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../data-exploration/filesets.html"> Fileset Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../data-exploration/tables.html"> Table Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../data-exploration/object-mapped-tables.html"> ObjectMappedTable Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../data-exploration/custom-datasets.html"> Custom Dataset Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../data-exploration/hive-execution-engines.html"> Hive Execution Engines</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../../advanced/index.html"> 高级主题</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../../advanced/application-logback.html"> Application Logback</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../advanced/best-practices.html"> Best Practices</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../advanced/class-loading.html"> Class Loading</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../advanced/configuring-resources.html"> Configuring Program Resources</a></li>
<li class="toctree-l2"><a class="reference internal" href="../../advanced/program-retry-policies.html"> Program Retry Policies</a></li>
</ul>
</li>
</ul>
</nav>
      </li>
      <li class="toctree-l1"><a href="../../table-of-contents/../../admin-manual/index.html" rel="nofollow">管理手册</a>
      </li>
      <li class="toctree-l1"><a href="../../table-of-contents/../../integrations/index.html" rel="nofollow">集成手册</a>
      </li>
      <li class="toctree-l1"><a href="../../table-of-contents/../../examples-manual/index.html" rel="nofollow">最佳实践</a>
      </li>
    </ul>
  </div></div>
    </div><div class="col-md-8 content" id="main-content">
    
  <div class="section" id="creating-a-plugin">
<span id="cdap-pipelines-creating-a-plugin"></span><h1>Creating a Plugin<a class="headerlink" href="#creating-a-plugin" title="Permalink to this headline">🔗</a></h1>
<div class="section" id="action-plugin">
<h2>Action Plugin<a class="headerlink" href="#action-plugin" title="Permalink to this headline">🔗</a></h2>
<p>An <code class="docutils literal notranslate"><span class="pre">Action</span></code> plugin runs arbitrary logic at the start or end of a batch data pipeline.</p>
<p>In order to implement an Action plugin, you
extend the <code class="docutils literal notranslate"><span class="pre">Action</span></code> class. Only one method is required to be implemented:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">run</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">run()</span></code>: Used to implement the functionality of the plugin.</li>
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Action that moves files from one fileset into another, optionally filtering files that match a regex.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">Action</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">FilesetMoveAction</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Action that moves files from one fileset into another, optionally filtering files that match a regex.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">FilesetMoveAction</span> <span class="kd">extends</span> <span class="n">Action</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;FilesetMove&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">SOURCE_FILESET</span> <span class="o">=</span> <span class="s">&quot;sourceFileset&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">DEST_FILESET</span> <span class="o">=</span> <span class="s">&quot;destinationFileset&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FILTER_REGEX</span> <span class="o">=</span> <span class="s">&quot;filterRegex&quot;</span><span class="p">;</span>

    <span class="nd">@Name</span><span class="p">(</span><span class="n">SOURCE_FILESET</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The fileset to move files from.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">sourceFileset</span><span class="p">;</span>

    <span class="nd">@Name</span><span class="p">(</span><span class="n">DEST_FILESET</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The fileset to move files to.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">destinationFileset</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">FILTER_REGEX</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Filter any files whose name matches this regex. Defaults to &#39;^\\.&#39;, which will filter any files &quot;</span> <span class="o">+</span>
      <span class="s">&quot;that begin with a period.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">filterRegex</span><span class="p">;</span>

    <span class="c1">// set defaults for properties in a no-argument constructor.</span>
    <span class="kd">public</span> <span class="nf">Conf</span><span class="p">()</span> <span class="p">{</span>
      <span class="n">filterRegex</span> <span class="o">=</span> <span class="s">&quot;^\\.&quot;</span><span class="p">;</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">FilesetMoveAction</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">filterRegex</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">ActionContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">context</span><span class="p">.</span><span class="na">execute</span><span class="p">(</span><span class="k">new</span> <span class="n">TxRunnable</span><span class="p">()</span> <span class="p">{</span>
      <span class="nd">@Override</span>
      <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">DatasetContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
        <span class="n">FileSet</span> <span class="n">sourceFileSet</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">sourceFileset</span><span class="p">);</span>
        <span class="n">FileSet</span> <span class="n">destinationFileSet</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">destinationFileset</span><span class="p">);</span>

        <span class="n">Pattern</span> <span class="n">pattern</span> <span class="o">=</span> <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">filterRegex</span><span class="p">);</span>

        <span class="k">for</span> <span class="p">(</span><span class="n">Location</span> <span class="n">sourceFile</span> <span class="p">:</span> <span class="n">sourceFileSet</span><span class="p">.</span><span class="na">getBaseLocation</span><span class="p">().</span><span class="na">list</span><span class="p">())</span> <span class="p">{</span>
          <span class="k">if</span> <span class="p">(</span><span class="n">pattern</span><span class="p">.</span><span class="na">matcher</span><span class="p">(</span><span class="n">sourceFile</span><span class="p">.</span><span class="na">getName</span><span class="p">()).</span><span class="na">find</span><span class="p">())</span> <span class="p">{</span>
            <span class="k">continue</span><span class="p">;</span>
          <span class="p">}</span>
          <span class="n">Location</span> <span class="n">destFile</span> <span class="o">=</span> <span class="n">destinationFileSet</span><span class="p">.</span><span class="na">getBaseLocation</span><span class="p">().</span><span class="na">append</span><span class="p">(</span><span class="n">sourceFile</span><span class="p">.</span><span class="na">getName</span><span class="p">());</span>
          <span class="n">sourceFile</span><span class="p">.</span><span class="na">renameTo</span><span class="p">(</span><span class="n">destFile</span><span class="p">);</span>
        <span class="p">}</span>
      <span class="p">}</span>
    <span class="p">});</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="post-run-action-plugin">
<h2>Post-run Action Plugin<a class="headerlink" href="#post-run-action-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">PostAction</span></code> plugin runs arbitrary logic after the end of a pipeline run.
It can be set to execute based on whether the run completed successfully,
if it failed, or in either case.</p>
<p>The difference between a <code class="docutils literal notranslate"><span class="pre">PostAction</span></code>—and an <code class="docutils literal notranslate"><span class="pre">Action</span></code> that is placed at the
end of a pipeline—is that a <code class="docutils literal notranslate"><span class="pre">PostAction</span></code> will always be executed even if the pipeline run fails,
while an <code class="docutils literal notranslate"><span class="pre">Action</span></code> will only be executed if every stage preceding it successfully runs.</p>
<p>In order to implement an Post-run Action plugin, you extend the <code class="docutils literal notranslate"><span class="pre">PostAction</span></code> class.
Only one method is required to be implemented:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">run</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">run()</span></code>: Used to implement the functionality of the plugin.</li>
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Post run action that deletes files in a FileSet that match a configurable regex.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">PostAction</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">FilesetDeletePostAction</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Post run action that deletes files in a FileSet that match a configurable regex if the run succeeded.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">FilesetDeletePostAction</span> <span class="kd">extends</span> <span class="n">PostAction</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;FilesetDelete&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FILESET_NAME</span> <span class="o">=</span> <span class="s">&quot;filesetName&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">DELETE_REGEX</span> <span class="o">=</span> <span class="s">&quot;deleteRegex&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">DIRECTORY</span> <span class="o">=</span> <span class="s">&quot;directory&quot;</span><span class="p">;</span>

    <span class="nd">@Name</span><span class="p">(</span><span class="n">FILESET_NAME</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The fileset to delete files from.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">filesetName</span><span class="p">;</span>

    <span class="nd">@Name</span><span class="p">(</span><span class="n">DELETE_REGEX</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Delete files that match this regex.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">deleteRegex</span><span class="p">;</span>

    <span class="c1">// Macro enabled properties can be set to a placeholder value ${key} when the pipeline is deployed.</span>
    <span class="c1">// At runtime, the value for &#39;key&#39; can be given and substituted in.</span>
    <span class="nd">@Macro</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">DIRECTORY</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The fileset directory to delete files from.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">directory</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">FilesetDeletePostAction</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">deleteRegex</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">BatchActionContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="k">if</span> <span class="p">(</span><span class="o">!</span><span class="n">context</span><span class="p">.</span><span class="na">isSuccessful</span><span class="p">())</span> <span class="p">{</span>
      <span class="k">return</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="n">FileSet</span> <span class="n">fileSet</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">filesetName</span><span class="p">);</span>
    <span class="n">Pattern</span> <span class="n">pattern</span> <span class="o">=</span> <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">deleteRegex</span><span class="p">);</span>
    <span class="k">for</span> <span class="p">(</span><span class="n">Location</span> <span class="n">fileLocation</span> <span class="p">:</span> <span class="n">fileSet</span><span class="p">.</span><span class="na">getBaseLocation</span><span class="p">().</span><span class="na">append</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">directory</span><span class="p">).</span><span class="na">list</span><span class="p">())</span> <span class="p">{</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">pattern</span><span class="p">.</span><span class="na">matcher</span><span class="p">(</span><span class="n">fileLocation</span><span class="p">.</span><span class="na">getName</span><span class="p">()).</span><span class="na">find</span><span class="p">())</span> <span class="p">{</span>
        <span class="n">fileLocation</span><span class="p">.</span><span class="na">delete</span><span class="p">();</span>
      <span class="p">}</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="batch-source-plugin">
<h2>Batch Source Plugin<a class="headerlink" href="#batch-source-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">BatchSource</span></code> plugin is used as a source of a batch data pipeline. It is used to prepare
and configure the input of a pipeline run.</p>
<p>In order to implement a Batch Source, you extend the
<code class="docutils literal notranslate"><span class="pre">BatchSource</span></code> class. You need to define the types of the <code class="docutils literal notranslate"><span class="pre">KEY</span></code> and <code class="docutils literal notranslate"><span class="pre">VALUE</span></code> that the Batch
Source will receive and the type of object that the Batch Source will emit to the
subsequent stage (which could be either a Transformation or a Batch Sink). After defining
the types, only one method is required to be implemented:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">prepareRun</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">prepareRun()</span></code>: Used to configure the input for each run of the pipeline.
If the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a stream or dataset is a macro, their creation will happen during this stage.
This is called by the client that will submit the job for the pipeline run.</li>
<li><code class="docutils literal notranslate"><span class="pre">onRunFinish()</span></code>: Used to run any required logic at the end of a pipeline run. This is called
by the client that submitted the job for the pipeline run.</li>
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Initialize the Batch Source. Guaranteed to be executed before any call
to the plugin’s <code class="docutils literal notranslate"><span class="pre">transform</span></code> method. This is called by each executor of the job. For example,
if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Destroy any resources created by <code class="docutils literal notranslate"><span class="pre">initialize</span></code>. Guaranteed to be executed after all calls
to the plugin’s <code class="docutils literal notranslate"><span class="pre">transform</span></code> method have been made. This is called by each executor of the job.
For example, if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">transform()</span></code>: This method will be called for every input key-value pair generated by
the batch job. By default, the value is emitted to the subsequent stage.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Batch Source that reads from a FileSet that has its data formatted as text.</span>
<span class="cm"> *</span>
<span class="cm"> * LongWritable is the first parameter because that is the key used by Hadoop&#39;s {@link TextInputFormat}.</span>
<span class="cm"> * Similarly, Text is the second parameter because that is the value used by Hadoop&#39;s {@link TextInputFormat}.</span>
<span class="cm"> * {@link StructuredRecord} is the third parameter because that is what the source will output.</span>
<span class="cm"> * All the plugins included with Hydrator operate on StructuredRecord.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">BatchSource</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">TextFileSetSource</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Reads from a FileSet that has its data formatted as text.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">TextFileSetSource</span> <span class="kd">extends</span> <span class="n">BatchSource</span><span class="o">&lt;</span><span class="n">LongWritable</span><span class="p">,</span> <span class="n">Text</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;TextFileSet&quot;</span><span class="p">;</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">Schema</span> <span class="n">OUTPUT_SCHEMA</span> <span class="o">=</span> <span class="n">Schema</span><span class="p">.</span><span class="na">recordOf</span><span class="p">(</span>
    <span class="s">&quot;textRecord&quot;</span><span class="p">,</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;position&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">LONG</span><span class="p">)),</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;text&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">))</span>
  <span class="p">);</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FILESET_NAME</span> <span class="o">=</span> <span class="s">&quot;fileSetName&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FILES</span> <span class="o">=</span> <span class="s">&quot;files&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">CREATE_IF_NOT_EXISTS</span> <span class="o">=</span> <span class="s">&quot;createIfNotExists&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">DELETE_INPUT_ON_SUCCESS</span> <span class="o">=</span> <span class="s">&quot;deleteInputOnSuccess&quot;</span><span class="p">;</span>

    <span class="c1">// The name annotation tells CDAP what the property name is. It is optional, and defaults to the variable name.</span>
    <span class="c1">// Note:  only primitives (including boxed types) and string are the types that are supported</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">FILESET_NAME</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the FileSet to read from.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">fileSetName</span><span class="p">;</span>

    <span class="c1">// Macro enabled properties can be set to a placeholder value ${key} when the pipeline is deployed.</span>
    <span class="c1">// At runtime, the value for &#39;key&#39; can be given and substituted in.</span>
    <span class="nd">@Macro</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">FILES</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;A comma separated list of files in the FileSet to read.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">files</span><span class="p">;</span>

    <span class="c1">// A nullable fields tells CDAP that this is an optional field.</span>
    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">CREATE_IF_NOT_EXISTS</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Whether to create the FileSet if it doesn&#39;t already exist. Defaults to false.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">Boolean</span> <span class="n">createIfNotExists</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">DELETE_INPUT_ON_SUCCESS</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Whether to delete the data read by the source after the run succeeds. Defaults to false.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">Boolean</span> <span class="n">deleteInputOnSuccess</span><span class="p">;</span>

    <span class="c1">// Use a no-args constructor to set field defaults.</span>
    <span class="kd">public</span> <span class="nf">Conf</span><span class="p">()</span> <span class="p">{</span>
      <span class="n">createIfNotExists</span> <span class="o">=</span> <span class="kc">false</span><span class="p">;</span>
      <span class="n">deleteInputOnSuccess</span> <span class="o">=</span> <span class="kc">false</span><span class="p">;</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="c1">// CDAP will pass in a config with its fields populated based on the configuration given when creating the pipeline.</span>
  <span class="kd">public</span> <span class="nf">TextFileSetSource</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="c1">// configurePipeline is called exactly once when the pipeline is being created.</span>
  <span class="c1">// Any static configuration should be performed here.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// if the user has set createIfNotExists to true, create the FileSet here.</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">createIfNotExists</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">createDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fileSetName</span><span class="p">,</span>
                                       <span class="n">FileSet</span><span class="p">.</span><span class="na">class</span><span class="p">,</span>
                                       <span class="n">FileSetProperties</span><span class="p">.</span><span class="na">builder</span><span class="p">()</span>
                                         <span class="p">.</span><span class="na">setInputFormat</span><span class="p">(</span><span class="n">TextInputFormat</span><span class="p">.</span><span class="na">class</span><span class="p">)</span>
                                         <span class="p">.</span><span class="na">setOutputFormat</span><span class="p">(</span><span class="n">TextOutputFormat</span><span class="p">.</span><span class="na">class</span><span class="p">)</span>
                                         <span class="p">.</span><span class="na">setEnableExploreOnCreate</span><span class="p">(</span><span class="kc">true</span><span class="p">)</span>
                                         <span class="p">.</span><span class="na">setExploreFormat</span><span class="p">(</span><span class="s">&quot;text&quot;</span><span class="p">)</span>
                                         <span class="p">.</span><span class="na">setExploreSchema</span><span class="p">(</span><span class="s">&quot;text string&quot;</span><span class="p">)</span>
                                         <span class="p">.</span><span class="na">build</span><span class="p">()</span>
      <span class="p">);</span>
    <span class="p">}</span>
    <span class="c1">// set the output schema of this stage so that stages further down the pipeline will know their input schema.</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="c1">// prepareRun is called before every pipeline run, and is used to configure what the input should be,</span>
  <span class="c1">// as well as any arguments the input should use. It is called by the client that is submitting the batch job.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">prepareRun</span><span class="p">(</span><span class="n">BatchSourceContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">IOException</span> <span class="p">{</span>
    <span class="n">Map</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">arguments</span> <span class="o">=</span> <span class="k">new</span> <span class="n">HashMap</span><span class="o">&lt;&gt;</span><span class="p">();</span>
    <span class="n">FileSetArguments</span><span class="p">.</span><span class="na">setInputPaths</span><span class="p">(</span><span class="n">arguments</span><span class="p">,</span> <span class="n">config</span><span class="p">.</span><span class="na">files</span><span class="p">);</span>
    <span class="n">context</span><span class="p">.</span><span class="na">setInput</span><span class="p">(</span><span class="n">Input</span><span class="p">.</span><span class="na">ofDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fileSetName</span><span class="p">,</span> <span class="n">arguments</span><span class="p">));</span>
  <span class="p">}</span>

  <span class="c1">// onRunFinish is called at the end of the pipeline run by the client that submitted the batch job.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onRunFinish</span><span class="p">(</span><span class="kt">boolean</span> <span class="n">succeeded</span><span class="p">,</span> <span class="n">BatchSourceContext</span> <span class="n">context</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// perform any actions that should happen at the end of the run.</span>
    <span class="c1">// in our case, we want to delete the data read during this run if the run succeeded.</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">succeeded</span> <span class="o">&amp;&amp;</span> <span class="n">config</span><span class="p">.</span><span class="na">deleteInputOnSuccess</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">Map</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">arguments</span> <span class="o">=</span> <span class="k">new</span> <span class="n">HashMap</span><span class="o">&lt;&gt;</span><span class="p">();</span>
      <span class="n">FileSetArguments</span><span class="p">.</span><span class="na">setInputPaths</span><span class="p">(</span><span class="n">arguments</span><span class="p">,</span> <span class="n">config</span><span class="p">.</span><span class="na">files</span><span class="p">);</span>
      <span class="n">FileSet</span> <span class="n">fileSet</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fileSetName</span><span class="p">,</span> <span class="n">arguments</span><span class="p">);</span>
      <span class="k">for</span> <span class="p">(</span><span class="n">Location</span> <span class="n">inputLocation</span> <span class="p">:</span> <span class="n">fileSet</span><span class="p">.</span><span class="na">getInputLocations</span><span class="p">())</span> <span class="p">{</span>
        <span class="k">try</span> <span class="p">{</span>
          <span class="n">inputLocation</span><span class="p">.</span><span class="na">delete</span><span class="p">(</span><span class="kc">true</span><span class="p">);</span>
        <span class="p">}</span> <span class="k">catch</span> <span class="p">(</span><span class="n">IOException</span> <span class="n">e</span><span class="p">)</span> <span class="p">{</span>
          <span class="k">throw</span> <span class="k">new</span> <span class="n">RuntimeException</span><span class="p">(</span><span class="n">e</span><span class="p">);</span>
        <span class="p">}</span>
      <span class="p">}</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="c1">// initialize is called by each job executor before any call to transform is made.</span>
  <span class="c1">// This occurs at the start of the batch job run, after the job has been successfully submitted.</span>
  <span class="c1">// For example, if mapreduce is the execution engine, each mapper will call initialize at the start of the program.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">initialize</span><span class="p">(</span><span class="n">BatchRuntimeContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="kd">super</span><span class="p">.</span><span class="na">initialize</span><span class="p">(</span><span class="n">context</span><span class="p">);</span>
    <span class="c1">// create any resources required by transform()</span>
  <span class="p">}</span>

  <span class="c1">// destroy is called by each job executor at the end of its life.</span>
  <span class="c1">// For example, if mapreduce is the execution engine, each mapper will call destroy at the end of the program.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">destroy</span><span class="p">()</span> <span class="p">{</span>
    <span class="c1">// clean up any resources created by initialize</span>
  <span class="p">}</span>

  <span class="c1">// transform is used to transform the key-value pair output by the input into objects output by this source.</span>
  <span class="c1">// The output should be a StructuredRecord if you want the source to be compatible with the plugins included</span>
  <span class="c1">// with Hydrator.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">transform</span><span class="p">(</span><span class="n">KeyValue</span><span class="o">&lt;</span><span class="n">LongWritable</span><span class="p">,</span> <span class="n">Text</span><span class="o">&gt;</span> <span class="n">input</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">emitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">StructuredRecord</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">)</span>
                   <span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;position&quot;</span><span class="p">,</span> <span class="n">input</span><span class="p">.</span><span class="na">getKey</span><span class="p">().</span><span class="na">get</span><span class="p">())</span>
                   <span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;text&quot;</span><span class="p">,</span> <span class="n">input</span><span class="p">.</span><span class="na">getValue</span><span class="p">().</span><span class="na">toString</span><span class="p">())</span>
                   <span class="p">.</span><span class="na">build</span><span class="p">()</span>
    <span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p class="rubric">Lineage</p>
<p>For plugins that fetch data from non-CDAP sources, the lineage is registered using the <code class="docutils literal notranslate"><span class="pre">inputName</span></code> provided
when <code class="docutils literal notranslate"><span class="pre">setInput()</span></code> is invoked on <code class="docutils literal notranslate"><span class="pre">BatchSourceContext</span></code> in <code class="docutils literal notranslate"><span class="pre">prepareRun()</span></code>. Note that the <code class="docutils literal notranslate"><span class="pre">inputName</span></code> should
be a valid <code class="docutils literal notranslate"><span class="pre">DatasetId</span></code>. For example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="nd">@Override</span>
<span class="kd">public</span> <span class="kt">void</span> <span class="nf">prepareRun</span><span class="p">(</span><span class="n">BatchSourceContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
  <span class="p">...</span>
  <span class="n">context</span><span class="p">.</span><span class="na">setInput</span><span class="p">(</span><span class="n">Input</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;myExternalSource&quot;</span><span class="p">,</span> <span class="n">myInputFormatProvider</span><span class="p">));</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Lineage will be tracked using <code class="docutils literal notranslate"><span class="pre">myExternalSource</span></code>.</p>
</div>
<div class="section" id="batch-sink-plugin">
<h2>Batch Sink Plugin<a class="headerlink" href="#batch-sink-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">BatchSink</span></code> plugin is used to write data in either batch or real-time data pipelines.
It is used to prepare and configure the output of a batch of data from a pipeline run.</p>
<p>In order to implement a Batch Sink, you extend the
<code class="docutils literal notranslate"><span class="pre">BatchSink</span></code> class. Similar to a Batch Source, you need to define the types of the <code class="docutils literal notranslate"><span class="pre">KEY</span></code> and
<code class="docutils literal notranslate"><span class="pre">VALUE</span></code> that the Batch Sink will write in the Batch job and the type of object that it will
accept from the previous stage (which could be either a Transformation or a Batch Source).</p>
<p>After defining the types, only one method is required to be implemented:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">prepareRun</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">prepareRun()</span></code>: Used to configure the output for each run of the pipeline. This is called by
the client that will submit the job for the pipeline run.</li>
<li><code class="docutils literal notranslate"><span class="pre">onRunFinish()</span></code>: Used to run any required logic at the end of a pipeline run. This is called
by the client that submitted the job for the pipeline run.</li>
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Initialize the Batch Sink. Guaranteed to be executed before any call
to the plugin’s <code class="docutils literal notranslate"><span class="pre">transform</span></code> method. This is called by each executor of the job. For example,
if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Destroy any resources created by <code class="docutils literal notranslate"><span class="pre">initialize</span></code>. Guaranteed to be executed after all calls
to the plugin’s <code class="docutils literal notranslate"><span class="pre">transform</span></code> method have been made. This is called by each executor of the job.
For example, if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">transform()</span></code>: This method will be called for every object that is received from the
previous stage. The logic inside the method will transform the object to the key-value
pair expected by the Batch Sink’s output format. If you don’t override this method, the
incoming object is set as the key and the value is set to null.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Batch Sink that writes to a FileSet in text format.</span>
<span class="cm"> * Each record will be written as a single line, with record fields separated by a configurable separator.</span>
<span class="cm"> *</span>
<span class="cm"> * StructuredRecord is the first parameter because that is the input to the sink.</span>
<span class="cm"> * The second and third parameters are the key and value expected by Hadoop&#39;s {@link TextOutputFormat}.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">BatchSink</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">TextFileSetSink</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Writes to a FileSet in text format.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">TextFileSetSink</span> <span class="kd">extends</span> <span class="n">BatchSink</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="p">,</span> <span class="n">NullWritable</span><span class="p">,</span> <span class="n">Text</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;TextFileSet&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FILESET_NAME</span> <span class="o">=</span> <span class="s">&quot;fileSetName&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">OUTPUT_DIR</span> <span class="o">=</span> <span class="s">&quot;outputDir&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">FIELD_SEPARATOR</span> <span class="o">=</span> <span class="s">&quot;fieldSeparator&quot;</span><span class="p">;</span>

    <span class="c1">// The name annotation tells CDAP what the property name is. It is optional, and defaults to the variable name.</span>
    <span class="c1">// Note:  only primitives (including boxed types) and string are the types that are supported</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">FILESET_NAME</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the FileSet to write to.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">fileSetName</span><span class="p">;</span>

    <span class="c1">// Macro enabled properties can be set to a placeholder value ${key} when the pipeline is deployed.</span>
    <span class="c1">// At runtime, the value for &#39;key&#39; can be given and substituted in.</span>
    <span class="nd">@Macro</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">OUTPUT_DIR</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The FileSet directory to write to.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">outputDir</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">FIELD_SEPARATOR</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The separator to use to join input record fields together. Defaults to &#39;,&#39;.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">fieldSeparator</span><span class="p">;</span>

    <span class="c1">// Use a no-args constructor to set field defaults.</span>
    <span class="kd">public</span> <span class="nf">Conf</span><span class="p">()</span> <span class="p">{</span>
      <span class="n">fileSetName</span> <span class="o">=</span> <span class="s">&quot;&quot;</span><span class="p">;</span>
      <span class="n">fieldSeparator</span> <span class="o">=</span> <span class="s">&quot;,&quot;</span><span class="p">;</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="c1">// CDAP will pass in a config with its fields populated based on the configuration given when creating the pipeline.</span>
  <span class="kd">public</span> <span class="nf">TextFileSetSink</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="c1">// configurePipeline is called exactly once when the pipeline is being created.</span>
  <span class="c1">// Any static configuration should be performed here.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// create the FileSet here.</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">createDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fileSetName</span><span class="p">,</span>
                                     <span class="n">FileSet</span><span class="p">.</span><span class="na">class</span><span class="p">,</span>
                                     <span class="n">FileSetProperties</span><span class="p">.</span><span class="na">builder</span><span class="p">()</span>
                                       <span class="p">.</span><span class="na">setInputFormat</span><span class="p">(</span><span class="n">TextInputFormat</span><span class="p">.</span><span class="na">class</span><span class="p">)</span>
                                       <span class="p">.</span><span class="na">setOutputFormat</span><span class="p">(</span><span class="n">TextOutputFormat</span><span class="p">.</span><span class="na">class</span><span class="p">)</span>
                                       <span class="p">.</span><span class="na">setEnableExploreOnCreate</span><span class="p">(</span><span class="kc">true</span><span class="p">)</span>
                                       <span class="p">.</span><span class="na">setExploreFormat</span><span class="p">(</span><span class="s">&quot;text&quot;</span><span class="p">)</span>
                                       <span class="p">.</span><span class="na">setExploreSchema</span><span class="p">(</span><span class="s">&quot;text string&quot;</span><span class="p">)</span>
                                       <span class="p">.</span><span class="na">build</span><span class="p">()</span>
    <span class="p">);</span>
  <span class="p">}</span>

  <span class="c1">// prepareRun is called before every pipeline run, and is used to configure what the input should be,</span>
  <span class="c1">// as well as any arguments the input should use. It is called by the client that is submitting the batch job.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">prepareRun</span><span class="p">(</span><span class="n">BatchSinkContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">Map</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">arguments</span> <span class="o">=</span> <span class="k">new</span> <span class="n">HashMap</span><span class="o">&lt;&gt;</span><span class="p">();</span>
    <span class="n">FileSetArguments</span><span class="p">.</span><span class="na">setOutputPath</span><span class="p">(</span><span class="n">arguments</span><span class="p">,</span> <span class="n">config</span><span class="p">.</span><span class="na">outputDir</span><span class="p">);</span>
    <span class="n">context</span><span class="p">.</span><span class="na">addOutput</span><span class="p">(</span><span class="n">Output</span><span class="p">.</span><span class="na">ofDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fileSetName</span><span class="p">,</span> <span class="n">arguments</span><span class="p">));</span>
  <span class="p">}</span>

  <span class="c1">// onRunFinish is called at the end of the pipeline run by the client that submitted the batch job.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onRunFinish</span><span class="p">(</span><span class="kt">boolean</span> <span class="n">succeeded</span><span class="p">,</span> <span class="n">BatchSinkContext</span> <span class="n">context</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// perform any actions that should happen at the end of the run.</span>
  <span class="p">}</span>

  <span class="c1">// initialize is called by each job executor before any call to transform is made.</span>
  <span class="c1">// This occurs at the start of the batch job run, after the job has been successfully submitted.</span>
  <span class="c1">// For example, if mapreduce is the execution engine, each mapper will call initialize at the start of the program.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">initialize</span><span class="p">(</span><span class="n">BatchRuntimeContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="kd">super</span><span class="p">.</span><span class="na">initialize</span><span class="p">(</span><span class="n">context</span><span class="p">);</span>
    <span class="c1">// create any resources required by transform()</span>
  <span class="p">}</span>

  <span class="c1">// destroy is called by each job executor at the end of its life.</span>
  <span class="c1">// For example, if mapreduce is the execution engine, each mapper will call destroy at the end of the program.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">destroy</span><span class="p">()</span> <span class="p">{</span>
    <span class="c1">// clean up any resources created by initialize</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">transform</span><span class="p">(</span><span class="n">StructuredRecord</span> <span class="n">input</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">KeyValue</span><span class="o">&lt;</span><span class="n">NullWritable</span><span class="p">,</span> <span class="n">Text</span><span class="o">&gt;&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">StringBuilder</span> <span class="n">joinedFields</span> <span class="o">=</span> <span class="k">new</span> <span class="n">StringBuilder</span><span class="p">();</span>
    <span class="n">Iterator</span><span class="o">&lt;</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="o">&gt;</span> <span class="n">fieldIter</span> <span class="o">=</span> <span class="n">input</span><span class="p">.</span><span class="na">getSchema</span><span class="p">().</span><span class="na">getFields</span><span class="p">().</span><span class="na">iterator</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="o">!</span><span class="n">fieldIter</span><span class="p">.</span><span class="na">hasNext</span><span class="p">())</span> <span class="p">{</span>
      <span class="c1">// shouldn&#39;t happen</span>
      <span class="k">return</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="n">Object</span> <span class="n">val</span> <span class="o">=</span> <span class="n">input</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldIter</span><span class="p">.</span><span class="na">next</span><span class="p">().</span><span class="na">getName</span><span class="p">());</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">val</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">joinedFields</span><span class="p">.</span><span class="na">append</span><span class="p">(</span><span class="n">val</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="k">while</span> <span class="p">(</span><span class="n">fieldIter</span><span class="p">.</span><span class="na">hasNext</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">String</span> <span class="n">fieldName</span> <span class="o">=</span> <span class="n">fieldIter</span><span class="p">.</span><span class="na">next</span><span class="p">().</span><span class="na">getName</span><span class="p">();</span>
      <span class="n">joinedFields</span><span class="p">.</span><span class="na">append</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">fieldSeparator</span><span class="p">);</span>
      <span class="n">val</span> <span class="o">=</span> <span class="n">input</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldName</span><span class="p">);</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">val</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">joinedFields</span><span class="p">.</span><span class="na">append</span><span class="p">(</span><span class="n">val</span><span class="p">);</span>
      <span class="p">}</span>
    <span class="p">}</span>
    <span class="n">emitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="k">new</span> <span class="n">KeyValue</span><span class="o">&lt;&gt;</span><span class="p">(</span><span class="n">NullWritable</span><span class="p">.</span><span class="na">get</span><span class="p">(),</span> <span class="k">new</span> <span class="n">Text</span><span class="p">(</span><span class="n">joinedFields</span><span class="p">.</span><span class="na">toString</span><span class="p">())));</span>
  <span class="p">}</span>

<span class="p">}</span>
</pre></div>
</div>
<p class="rubric">Lineage</p>
<p>For plugins that write data to non-CDAP sinks, the lineage is registered using the <code class="docutils literal notranslate"><span class="pre">outputName</span></code> provided
when <code class="docutils literal notranslate"><span class="pre">addOutput()</span></code> is invoked on <code class="docutils literal notranslate"><span class="pre">BatchSinkContext</span></code> in <code class="docutils literal notranslate"><span class="pre">prepareRun()</span></code>. Note that the <code class="docutils literal notranslate"><span class="pre">outputName</span></code> should
be a valid <code class="docutils literal notranslate"><span class="pre">DatasetId</span></code>. For example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="nd">@Override</span>
<span class="kd">public</span> <span class="kt">void</span> <span class="nf">prepareRun</span><span class="p">(</span><span class="n">BatchSinkContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
  <span class="p">...</span>
  <span class="n">context</span><span class="p">.</span><span class="na">addOutput</span><span class="p">(</span><span class="n">Output</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;myExternalSink&quot;</span><span class="p">,</span> <span class="n">myOutputFormatProvider</span><span class="p">));</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Lineage will be tracked using <code class="docutils literal notranslate"><span class="pre">myExternalSink</span></code>.</p>
</div>
<div class="section" id="transformation-plugin">
<h2>Transformation Plugin<a class="headerlink" href="#transformation-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">Transform</span></code> plugin is used to convert one input record into zero or more output records.
It can be used in both batch and real-time data pipelines.</p>
<p>The only method that needs to be implemented is:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">transform</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Used to perform any initialization step that might be required during
the runtime of the <code class="docutils literal notranslate"><span class="pre">Transform</span></code>. It is guaranteed that this method will be invoked
before the <code class="docutils literal notranslate"><span class="pre">transform</span></code> method.</li>
<li><code class="docutils literal notranslate"><span class="pre">transform()</span></code>: This method contains the logic that will be applied on each
incoming data object. An emitter can be used to pass the results to the subsequent stage.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Used to perform any cleanup before the plugin shuts down.</li>
</ul>
<p>Below is an example of a <code class="docutils literal notranslate"><span class="pre">StringCase</span></code> that transforms specific fields to lowercase or uppercase.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Transform that can transforms specific fields to lowercase or uppercase.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">Transform</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">StringCaseTransform</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Transforms configured fields to lowercase or uppercase.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">StringCaseTransform</span> <span class="kd">extends</span> <span class="n">Transform</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;StringCase&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>
  <span class="kd">private</span> <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">upperFields</span><span class="p">;</span>
  <span class="kd">private</span> <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">lowerFields</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">UPPER_FIELDS</span> <span class="o">=</span> <span class="s">&quot;upperFields&quot;</span><span class="p">;</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">LOWER_FIELDS</span> <span class="o">=</span> <span class="s">&quot;lowerFields&quot;</span><span class="p">;</span>
    <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">Pattern</span> <span class="n">SPLIT_ON</span> <span class="o">=</span> <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="s">&quot;\\s*,\\s*&quot;</span><span class="p">);</span>

    <span class="c1">// nullable means this property is optional</span>
    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">UPPER_FIELDS</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;A comma separated list of fields to uppercase. Each field must be of type String.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">upperFields</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Name</span><span class="p">(</span><span class="n">LOWER_FIELDS</span><span class="p">)</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;A comma separated list of fields to lowercase. Each field must be of type String.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">lowerFields</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="nf">getUpperFields</span><span class="p">()</span> <span class="p">{</span>
      <span class="k">return</span> <span class="n">parseToSet</span><span class="p">(</span><span class="n">upperFields</span><span class="p">);</span>
    <span class="p">}</span>

    <span class="kd">private</span> <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="nf">getLowerFields</span><span class="p">()</span> <span class="p">{</span>
      <span class="k">return</span> <span class="n">parseToSet</span><span class="p">(</span><span class="n">lowerFields</span><span class="p">);</span>
    <span class="p">}</span>

    <span class="kd">private</span> <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="nf">parseToSet</span><span class="p">(</span><span class="n">String</span> <span class="n">str</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">Set</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">set</span> <span class="o">=</span> <span class="k">new</span> <span class="n">HashSet</span><span class="o">&lt;&gt;</span><span class="p">();</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">str</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">||</span> <span class="n">str</span><span class="p">.</span><span class="na">isEmpty</span><span class="p">())</span> <span class="p">{</span>
        <span class="k">return</span> <span class="n">set</span><span class="p">;</span>
      <span class="p">}</span>
      <span class="k">for</span> <span class="p">(</span><span class="n">String</span> <span class="n">element</span> <span class="p">:</span> <span class="n">SPLIT_ON</span><span class="p">.</span><span class="na">split</span><span class="p">(</span><span class="n">str</span><span class="p">))</span> <span class="p">{</span>
        <span class="n">set</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">element</span><span class="p">);</span>
      <span class="p">}</span>
      <span class="k">return</span> <span class="n">set</span><span class="p">;</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">StringCaseTransform</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="c1">// configurePipeline is called only once, when the pipeline is deployed. Static validation should be done here.</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">StageConfigurer</span> <span class="n">stageConfigurer</span> <span class="o">=</span> <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">();</span>
    <span class="c1">// the output schema is always the same as the input schema</span>
    <span class="n">Schema</span> <span class="n">inputSchema</span> <span class="o">=</span> <span class="n">stageConfigurer</span><span class="p">.</span><span class="na">getInputSchema</span><span class="p">();</span>

    <span class="c1">// if schema is null, that means it is either not known until runtime, or it is variable</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="c1">// if the input schema is constant and known at configure time, check that all configured fields are strings</span>
      <span class="k">for</span> <span class="p">(</span><span class="n">String</span> <span class="n">fieldName</span> <span class="p">:</span> <span class="n">config</span><span class="p">.</span><span class="na">getUpperFields</span><span class="p">())</span> <span class="p">{</span>
        <span class="n">validateFieldIsString</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">,</span> <span class="n">fieldName</span><span class="p">);</span>
      <span class="p">}</span>
      <span class="k">for</span> <span class="p">(</span><span class="n">String</span> <span class="n">fieldName</span> <span class="p">:</span> <span class="n">config</span><span class="p">.</span><span class="na">getLowerFields</span><span class="p">())</span> <span class="p">{</span>
        <span class="n">validateFieldIsString</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">,</span> <span class="n">fieldName</span><span class="p">);</span>
      <span class="p">}</span>
    <span class="p">}</span>

    <span class="n">stageConfigurer</span><span class="p">.</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="c1">// initialize is called once at the start of each pipeline run</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">initialize</span><span class="p">(</span><span class="n">TransformContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">upperFields</span> <span class="o">=</span> <span class="n">config</span><span class="p">.</span><span class="na">getUpperFields</span><span class="p">();</span>
    <span class="n">lowerFields</span> <span class="o">=</span> <span class="n">config</span><span class="p">.</span><span class="na">getLowerFields</span><span class="p">();</span>
  <span class="p">}</span>

  <span class="c1">// transform is called once for each record that goes into this stage</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">transform</span><span class="p">(</span><span class="n">StructuredRecord</span> <span class="n">record</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">StructuredRecord</span><span class="p">.</span><span class="na">Builder</span> <span class="n">builder</span> <span class="o">=</span> <span class="n">StructuredRecord</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="n">record</span><span class="p">.</span><span class="na">getSchema</span><span class="p">());</span>
    <span class="k">for</span> <span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span> <span class="n">field</span> <span class="p">:</span> <span class="n">record</span><span class="p">.</span><span class="na">getSchema</span><span class="p">().</span><span class="na">getFields</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">String</span> <span class="n">fieldName</span> <span class="o">=</span> <span class="n">field</span><span class="p">.</span><span class="na">getName</span><span class="p">();</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">upperFields</span><span class="p">.</span><span class="na">contains</span><span class="p">(</span><span class="n">fieldName</span><span class="p">))</span> <span class="p">{</span>
        <span class="n">builder</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">fieldName</span><span class="p">,</span> <span class="n">record</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldName</span><span class="p">).</span><span class="na">toString</span><span class="p">().</span><span class="na">toUpperCase</span><span class="p">());</span>
      <span class="p">}</span> <span class="k">else</span> <span class="k">if</span> <span class="p">(</span><span class="n">lowerFields</span><span class="p">.</span><span class="na">contains</span><span class="p">(</span><span class="n">fieldName</span><span class="p">))</span> <span class="p">{</span>
        <span class="n">builder</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">fieldName</span><span class="p">,</span> <span class="n">record</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldName</span><span class="p">).</span><span class="na">toString</span><span class="p">().</span><span class="na">toLowerCase</span><span class="p">());</span>
      <span class="p">}</span> <span class="k">else</span> <span class="p">{</span>
        <span class="n">builder</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">fieldName</span><span class="p">,</span> <span class="n">record</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldName</span><span class="p">));</span>
      <span class="p">}</span>
    <span class="p">}</span>
    <span class="n">emitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">builder</span><span class="p">.</span><span class="na">build</span><span class="p">());</span>
  <span class="p">}</span>

  <span class="kd">private</span> <span class="kt">void</span> <span class="nf">validateFieldIsString</span><span class="p">(</span><span class="n">Schema</span> <span class="n">schema</span><span class="p">,</span> <span class="n">String</span> <span class="n">fieldName</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span> <span class="n">inputField</span> <span class="o">=</span> <span class="n">schema</span><span class="p">.</span><span class="na">getField</span><span class="p">(</span><span class="n">fieldName</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputField</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span>
        <span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="s">&quot;Field &#39;%s&#39; does not exist in input schema %s.&quot;</span><span class="p">,</span> <span class="n">fieldName</span><span class="p">,</span> <span class="n">schema</span><span class="p">));</span>
    <span class="p">}</span>
    <span class="n">Schema</span> <span class="n">fieldSchema</span> <span class="o">=</span> <span class="n">inputField</span><span class="p">.</span><span class="na">getSchema</span><span class="p">();</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span> <span class="n">fieldType</span> <span class="o">=</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">isNullable</span><span class="p">()</span> <span class="o">?</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">getNonNullable</span><span class="p">().</span><span class="na">getType</span><span class="p">()</span> <span class="p">:</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">getType</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">fieldType</span> <span class="o">!=</span> <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span>
        <span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="s">&quot;Field &#39;%s&#39; is of illegal type %s. Must be of type %s.&quot;</span><span class="p">,</span>
                      <span class="n">fieldName</span><span class="p">,</span> <span class="n">fieldType</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">));</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>If you wanted, you could add to the <code class="docutils literal notranslate"><span class="pre">transform</span></code> method a user metric indicating the
number of fields changed. The user metrics can be queried by using the CDAP
<span class="xref std std-ref">Metrics HTTP RESTful API</span>:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kt">void</span> <span class="nf">transform</span><span class="p">(</span><span class="n">StructuredRecord</span> <span class="n">input</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
  <span class="kt">int</span> <span class="n">fieldsChanged</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
  <span class="p">.</span> <span class="p">.</span> <span class="p">.</span>
    <span class="n">builder</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">fieldName</span><span class="p">,</span> <span class="n">record</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">fieldName</span><span class="p">).</span> <span class="p">.</span> <span class="p">.</span>
    <span class="n">fieldsChanged</span> <span class="o">+=</span> <span class="mi">1</span><span class="p">;</span>
  <span class="p">.</span> <span class="p">.</span> <span class="p">.</span>
  <span class="n">getContext</span><span class="p">().</span><span class="na">getMetrics</span><span class="p">().</span><span class="na">count</span><span class="p">(</span><span class="s">&quot;fieldsChanged&quot;</span><span class="p">,</span> <span class="n">fieldsChanged</span><span class="p">);</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="error-transformation-plugin">
<h2>Error Transformation Plugin<a class="headerlink" href="#error-transformation-plugin" title="Permalink to this headline">🔗</a></h2>
<p>An <code class="docutils literal notranslate"><span class="pre">ErrorTransform</span></code> plugin is a special type of <code class="docutils literal notranslate"><span class="pre">Transform</span></code> that consumes error records emitted
from the previous stages instead of output records. It is used to transform an <code class="docutils literal notranslate"><span class="pre">ErrorRecord</span></code> to zero
or more output records. In addition to the actual error object, an <code class="docutils literal notranslate"><span class="pre">ErrorRecord</span></code> exposes the stage the
error was emitted from, an error code, and an error message. Errors can be emitted by <code class="docutils literal notranslate"><span class="pre">BatchSource</span></code>,
<code class="docutils literal notranslate"><span class="pre">Transform</span></code>, and <code class="docutils literal notranslate"><span class="pre">BatchAggregator</span></code> plugins using the <code class="docutils literal notranslate"><span class="pre">Emitter</span></code> they receive.
An <code class="docutils literal notranslate"><span class="pre">ErrorTransform</span></code> can be used in both batch and real-time data pipelines.</p>
<p>The only method that needs to be implemented is:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">transform</span><span class="p">()</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Used to perform any initialization step that might be required during
the runtime of the <code class="docutils literal notranslate"><span class="pre">ErrorTransform</span></code>. It is guaranteed that this method will be invoked
before the <code class="docutils literal notranslate"><span class="pre">transform</span></code> method.</li>
<li><code class="docutils literal notranslate"><span class="pre">transform()</span></code>: This method contains the logic that will be applied on each
incoming <code class="docutils literal notranslate"><span class="pre">ErrorRecord</span></code> object. An emitter can be used to pass the results to the subsequent stage.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Used to perform any cleanup before the plugin shuts down.</li>
</ul>
<p>Below is an example of an <code class="docutils literal notranslate"><span class="pre">ErrorCollector</span></code> that adds the error stage, code, and message to each record it receives.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Adds the error code and error message to each record, then emits it.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">ErrorTransform</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="s">&quot;ErrorCollector&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">ErrorCollector</span> <span class="kd">extends</span> <span class="n">ErrorTransform</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Config</span> <span class="n">config</span><span class="p">;</span>

  <span class="kd">public</span> <span class="nf">ErrorCollector</span><span class="p">(</span><span class="n">Config</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">IllegalArgumentException</span> <span class="p">{</span>
    <span class="n">Schema</span> <span class="n">inputSchema</span> <span class="o">=</span> <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">getInputSchema</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span><span class="p">.</span><span class="na">getField</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">messageField</span><span class="p">)</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span><span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span>
          <span class="s">&quot;Input schema already contains message field %s. Please set messageField to a different value.&quot;</span><span class="p">,</span>
          <span class="n">config</span><span class="p">.</span><span class="na">messageField</span><span class="p">));</span>
      <span class="p">}</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span><span class="p">.</span><span class="na">getField</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">codeField</span><span class="p">)</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span><span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span>
          <span class="s">&quot;Input schema already contains code field %s. Please set codeField to a different value.&quot;</span><span class="p">,</span>
          <span class="n">config</span><span class="p">.</span><span class="na">codeField</span><span class="p">));</span>
      <span class="p">}</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span><span class="p">.</span><span class="na">getField</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">stageField</span><span class="p">)</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span><span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span>
          <span class="s">&quot;Input schema already contains stage field %s. Please set stageField to a different value.&quot;</span><span class="p">,</span>
          <span class="n">config</span><span class="p">.</span><span class="na">stageField</span><span class="p">));</span>
      <span class="p">}</span>
      <span class="n">Schema</span> <span class="n">outputSchema</span> <span class="o">=</span> <span class="n">getOutputSchema</span><span class="p">(</span><span class="n">config</span><span class="p">,</span> <span class="n">inputSchema</span><span class="p">);</span>
      <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">outputSchema</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">transform</span><span class="p">(</span><span class="n">ErrorRecord</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">input</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">StructuredRecord</span> <span class="n">invalidRecord</span> <span class="o">=</span> <span class="n">input</span><span class="p">.</span><span class="na">getRecord</span><span class="p">();</span>
    <span class="n">StructuredRecord</span><span class="p">.</span><span class="na">Builder</span> <span class="n">output</span> <span class="o">=</span> <span class="n">StructuredRecord</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="n">getOutputSchema</span><span class="p">(</span><span class="n">config</span><span class="p">,</span> <span class="n">invalidRecord</span><span class="p">.</span><span class="na">getSchema</span><span class="p">()));</span>
    <span class="k">for</span> <span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span> <span class="n">field</span> <span class="p">:</span> <span class="n">invalidRecord</span><span class="p">.</span><span class="na">getSchema</span><span class="p">().</span><span class="na">getFields</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">output</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">field</span><span class="p">.</span><span class="na">getName</span><span class="p">(),</span> <span class="n">invalidRecord</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">field</span><span class="p">.</span><span class="na">getName</span><span class="p">()));</span>
    <span class="p">}</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">messageField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">output</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">messageField</span><span class="p">,</span> <span class="n">input</span><span class="p">.</span><span class="na">getErrorMessage</span><span class="p">());</span>
    <span class="p">}</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">codeField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">output</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">codeField</span><span class="p">,</span> <span class="n">input</span><span class="p">.</span><span class="na">getErrorCode</span><span class="p">());</span>
    <span class="p">}</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">stageField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">output</span><span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">stageField</span><span class="p">,</span> <span class="n">input</span><span class="p">.</span><span class="na">getStageName</span><span class="p">());</span>
    <span class="p">}</span>
    <span class="n">emitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">output</span><span class="p">.</span><span class="na">build</span><span class="p">());</span>
  <span class="p">}</span>

  <span class="kd">private</span> <span class="kd">static</span> <span class="n">Schema</span> <span class="nf">getOutputSchema</span><span class="p">(</span><span class="n">Config</span> <span class="n">config</span><span class="p">,</span> <span class="n">Schema</span> <span class="n">inputSchema</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">List</span><span class="o">&lt;</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="o">&gt;</span> <span class="n">fields</span> <span class="o">=</span> <span class="k">new</span> <span class="n">ArrayList</span><span class="o">&lt;&gt;</span><span class="p">();</span>
    <span class="n">fields</span><span class="p">.</span><span class="na">addAll</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">.</span><span class="na">getFields</span><span class="p">());</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">messageField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">fields</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">messageField</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)));</span>
    <span class="p">}</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">codeField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">fields</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">codeField</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">INT</span><span class="p">)));</span>
    <span class="p">}</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">stageField</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">fields</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">stageField</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)));</span>
    <span class="p">}</span>
    <span class="k">return</span> <span class="n">Schema</span><span class="p">.</span><span class="na">recordOf</span><span class="p">(</span><span class="s">&quot;error&quot;</span> <span class="o">+</span> <span class="n">inputSchema</span><span class="p">.</span><span class="na">getRecordName</span><span class="p">(),</span> <span class="n">fields</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="cm">/**</span>
<span class="cm">   * The plugin config</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Config</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="nd">@Nullable</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the error message field to use in the output schema. &quot;</span> <span class="o">+</span>
      <span class="s">&quot;If this not specified, the error message will be dropped.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">messageField</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the error code field to use in the output schema. &quot;</span> <span class="o">+</span>
      <span class="s">&quot;If this not specified, the error code will be dropped.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">codeField</span><span class="p">;</span>

    <span class="nd">@Nullable</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the error stage field to use in the output schema. &quot;</span> <span class="o">+</span>
      <span class="s">&quot;If this not specified, the error stage will be dropped.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">stageField</span><span class="p">;</span>

  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="alert-publisher-plugin">
<h2>Alert Publisher Plugin<a class="headerlink" href="#alert-publisher-plugin" title="Permalink to this headline">🔗</a></h2>
<p>An <code class="docutils literal notranslate"><span class="pre">AlertPublisher</span></code> plugin is a special type of plugin that consumes alerts emitted from previous stages
instead of output records. Alerts are meant to be uncommon events that need to be acted on in some other
program. Alerts contain a payload, which is just a map of strings containing any relevant data.
An alert publisher is responsible for writing the alerts to some system, where it can be read and acted
upon by some external program. For example, a plugin may write alerts to Kafka.
Alerts may not be published immediately after they are emitted.
It is up to the processing engine to decide when to publish alerts.</p>
<p>The only method that needs to be implemented is:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">publish</span><span class="p">(</span><span class="n">Iterator</span><span class="o">&lt;</span><span class="n">Alert</span><span class="o">&gt;</span> <span class="n">alerts</span><span class="p">)</span>
</pre></div>
</div>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Used to perform any initialization step that might be required during
the runtime of the <code class="docutils literal notranslate"><span class="pre">AlertPublisher</span></code>. It is guaranteed that this method will be invoked
before the <code class="docutils literal notranslate"><span class="pre">publish</span></code> method.</li>
<li><code class="docutils literal notranslate"><span class="pre">publish()</span></code>: This method contains the logic that will publish each incoming Alert.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Used to perform any cleanup before the plugin shuts down.</li>
</ul>
</div>
<div class="section" id="script-transformations">
<h2>Script Transformations<a class="headerlink" href="#script-transformations" title="Permalink to this headline">🔗</a></h2>
<p>In the script transformations (<em>JavaScriptTransform</em>, <em>PythonEvaluator</em>,
and the <em>ValidatorTransform</em>), a <code class="docutils literal notranslate"><span class="pre">ScriptContext</span></code> object is
passed to the <code class="docutils literal notranslate"><span class="pre">transform()</span></code> method:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">function</span> <span class="nf">transform</span><span class="p">(</span><span class="n">input</span><span class="p">,</span> <span class="n">context</span><span class="p">);</span>
</pre></div>
</div>
<p>The different Transforms that are passed this context object have similar signatures:</p>
<table border="1" class="docutils">
<colgroup>
<col width="20%" />
<col width="80%" />
</colgroup>
<thead valign="bottom">
<tr class="row-odd"><th class="head">Transform</th>
<th class="head">Signature</th>
</tr>
</thead>
<tbody valign="top">
<tr class="row-even"><td><code class="docutils literal notranslate"><span class="pre">JavaScriptTransform</span></code></td>
<td><code class="docutils literal notranslate"><span class="pre">{{function</span> <span class="pre">transform(input,</span> <span class="pre">emitter,</span> <span class="pre">context)}}</span></code></td>
</tr>
<tr class="row-odd"><td><code class="docutils literal notranslate"><span class="pre">PythonEvaluator</span></code></td>
<td><code class="docutils literal notranslate"><span class="pre">{{function</span> <span class="pre">transform(input,</span> <span class="pre">emitter,</span> <span class="pre">context)}}</span></code></td>
</tr>
<tr class="row-even"><td><code class="docutils literal notranslate"><span class="pre">ValidatorTransform</span></code></td>
<td><code class="docutils literal notranslate"><span class="pre">{{function</span> <span class="pre">isValid(input,</span> <span class="pre">context)}}</span></code></td>
</tr>
</tbody>
</table>
<p>The <code class="docutils literal notranslate"><span class="pre">ScriptContext</span></code> has these methods:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="n">Logger</span> <span class="nf">getLogger</span><span class="p">();</span>
<span class="kd">public</span> <span class="n">StageMetrics</span> <span class="nf">getMetrics</span><span class="p">();</span>
<span class="kd">public</span> <span class="n">ScriptLookup</span> <span class="nf">getLookup</span><span class="p">(</span><span class="n">String</span> <span class="n">table</span><span class="p">);</span>
</pre></div>
</div>
<p>The context passed by the <em>ValidatorTransform</em> has an additional method that returns a validator:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="n">Object</span> <span class="nf">getValidator</span><span class="p">(</span><span class="n">String</span> <span class="n">validatorName</span><span class="p">);</span>
</pre></div>
</div>
<p>These methods allow access within the script to CDAP loggers, metrics, lookup tables, and the validator object.</p>
<p><strong>Logger</strong></p>
<p><code class="docutils literal notranslate"><span class="pre">Logger</span></code> is an <a class="reference external" href="http://www.slf4j.org/api/org/slf4j/Logger.html">org.slf4j.Logger</a>.</p>
<p>For example, a JavaScript transform step can access and write to the <em>debug</em> log with:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span>context.getLogger().debug(&#39;Received record with id &#39; + input.id);
</pre></div>
</div>
<p><strong>StageMetrics</strong></p>
<p><code class="docutils literal notranslate"><span class="pre">StageMetrics</span></code> has these methods:</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">count(String</span> <span class="pre">metricName,</span> <span class="pre">int</span> <span class="pre">delta)</span></code>: Increases the value of the specific metric by
delta. Metrics name will be prefixed by the stage ID, hence it will be aggregated for
the current stage.</li>
<li><code class="docutils literal notranslate"><span class="pre">gauge(String</span> <span class="pre">metricName,</span> <span class="pre">long</span> <span class="pre">value)</span></code>: Sets the specific metric to the provided
value. Metrics name will be prefixed by the stage ID, hence it will be aggregated for
the current stage.</li>
<li><code class="docutils literal notranslate"><span class="pre">pipelineCount(String</span> <span class="pre">metricName,</span> <span class="pre">int</span> <span class="pre">delta)</span></code>: Increases the value of the specific
metric by delta. Metrics emitted will be aggregated for the entire pipeline.</li>
<li><code class="docutils literal notranslate"><span class="pre">pipelineGauge(String</span> <span class="pre">metricName,</span> <span class="pre">long</span> <span class="pre">value)</span></code>: Sets the specific metric to the
provided value. Metrics emitted will be aggregated for the entire pipeline.</li>
</ul>
<p><strong>ScriptLookup</strong></p>
<p>Currently, <code class="docutils literal notranslate"><span class="pre">ScriptContext.getLookup(String</span> <span class="pre">table)</span></code> only supports <a class="reference internal" href="../../building-blocks/datasets/index.html#datasets-index"><span class="std std-ref">key-value tables</span></a>.</p>
<p>For example, if a lookup table <em>purchases</em> is configured, then you will be able to perform
operations with that lookup table in your script: <code class="docutils literal notranslate"><span class="pre">context.getLookup('purchases').lookup('key')</span></code></p>
<p><strong>Validator Object</strong></p>
<p>For example, in a validator transform, you can retrieve the validator object and call its
functions as part of your JavaScript:</p>
<div class="highlight-javascript notranslate"><div class="highlight"><pre><span></span><span class="kd">var</span> <span class="nx">coreValidator</span> <span class="o">=</span> <span class="nx">context</span><span class="p">.</span><span class="nx">getValidator</span><span class="p">(</span><span class="s2">&quot;coreValidator&quot;</span><span class="p">);</span>
<span class="k">if</span> <span class="p">(</span><span class="o">!</span><span class="nx">coreValidator</span><span class="p">.</span><span class="nx">isDate</span><span class="p">(</span><span class="nx">input</span><span class="p">.</span><span class="nx">date</span><span class="p">))</span> <span class="p">{</span>
<span class="p">.</span> <span class="p">.</span> <span class="p">.</span>
</pre></div>
</div>
</div>
<div class="section" id="batch-aggregator-plugin">
<h2>Batch Aggregator Plugin<a class="headerlink" href="#batch-aggregator-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">BatchAggregator</span></code> plugin is used to compute aggregates over a batch of data.
It is used in both batch and real-time data pipelines.
An aggregation takes place in two steps: <em>groupBy</em> and then <em>aggregate</em>.</p>
<ul class="simple">
<li>In the <em>groupBy</em> step, the aggregator creates zero or more group keys for each input
record. Before the <em>aggregate</em> step occurs, the CDAP pipeline will take all records that have the
same group key, and collect them into a group. If a record does not have any of the
group keys, it is filtered out. If a record has multiple group keys, it will belong to
multiple groups.</li>
<li>The <em>aggregate</em> step is then called. In this step, the plugin receives group keys and
all records that had that group key. It is then left to the plugin to decide what to do
with each of the groups.</li>
</ul>
<p>In order to implement a Batch Aggregator, you extend the
<code class="docutils literal notranslate"><span class="pre">BatchAggregator</span></code> class. Unlike a <code class="docutils literal notranslate"><span class="pre">Transform</span></code>, which operates on a single record at a time, a
<code class="docutils literal notranslate"><span class="pre">BatchAggregator</span></code> operates on a collection of records.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Initialize the Batch Aggregator. Guaranteed to be executed before any call
to the plugin’s <code class="docutils literal notranslate"><span class="pre">groupBy</span></code> or <code class="docutils literal notranslate"><span class="pre">aggregate</span></code> methods. This is called by each executor of the job.
For example, if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Destroy any resources created by <code class="docutils literal notranslate"><span class="pre">initialize</span></code>. Guaranteed to be
executed after all calls to the plugin’s <code class="docutils literal notranslate"><span class="pre">groupBy</span></code> or <code class="docutils literal notranslate"><span class="pre">aggregate</span></code> methods have been
made. This is called by each executor of the job. For example, if the MapReduce engine
is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">groupBy()</span></code>: This method will be called for every object that is received from the
previous stage. This method returns zero or more group keys for each object it recieves.
Objects with the same group key will be grouped together for the <code class="docutils literal notranslate"><span class="pre">aggregate</span></code> method.</li>
<li><code class="docutils literal notranslate"><span class="pre">aggregate()</span></code>: The method is called after every object has been assigned their group keys.
This method is called once for each group key emitted by the <code class="docutils literal notranslate"><span class="pre">groupBy</span></code> method.
The method recieves a group key as well as an iterator over all objects that had that group key.
Objects emitted in this method are the output for this stage.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * Aggregator that counts how many times each word appears in records input to the aggregator.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">BatchAggregator</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">WordCountAggregator</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Counts how many times each word appears in all records input to the aggregator.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">WordCountAggregator</span> <span class="kd">extends</span> <span class="n">BatchAggregator</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;WordCount&quot;</span><span class="p">;</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">Schema</span> <span class="n">OUTPUT_SCHEMA</span> <span class="o">=</span> <span class="n">Schema</span><span class="p">.</span><span class="na">recordOf</span><span class="p">(</span>
    <span class="s">&quot;wordCount&quot;</span><span class="p">,</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;word&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)),</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;count&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">LONG</span><span class="p">))</span>
  <span class="p">);</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">Pattern</span> <span class="n">WHITESPACE</span> <span class="o">=</span> <span class="n">Pattern</span><span class="p">.</span><span class="na">compile</span><span class="p">(</span><span class="s">&quot;\\s&quot;</span><span class="p">);</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The field from the input records containing the words to count.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">field</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">WordCountAggregator</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// any static configuration validation should happen here.</span>
    <span class="c1">// We will check that the field is in the input schema and is of type string.</span>
    <span class="n">Schema</span> <span class="n">inputSchema</span> <span class="o">=</span> <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">getInputSchema</span><span class="p">();</span>
    <span class="c1">// a null input schema means its unknown until runtime, or its not constant</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="c1">// if the input schema is constant and known at configure time, check that the input field exists and is a string.</span>
      <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span> <span class="n">inputField</span> <span class="o">=</span> <span class="n">inputSchema</span><span class="p">.</span><span class="na">getField</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">inputField</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span>
          <span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="s">&quot;Field &#39;%s&#39; does not exist in input schema %s.&quot;</span><span class="p">,</span> <span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">,</span> <span class="n">inputSchema</span><span class="p">));</span>
      <span class="p">}</span>
      <span class="n">Schema</span> <span class="n">fieldSchema</span> <span class="o">=</span> <span class="n">inputField</span><span class="p">.</span><span class="na">getSchema</span><span class="p">();</span>
      <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span> <span class="n">fieldType</span> <span class="o">=</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">isNullable</span><span class="p">()</span> <span class="o">?</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">getNonNullable</span><span class="p">().</span><span class="na">getType</span><span class="p">()</span> <span class="p">:</span> <span class="n">fieldSchema</span><span class="p">.</span><span class="na">getType</span><span class="p">();</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">fieldType</span> <span class="o">!=</span> <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="n">IllegalArgumentException</span><span class="p">(</span>
          <span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="s">&quot;Field &#39;%s&#39; is of illegal type %s. Must be of type %s.&quot;</span><span class="p">,</span>
                        <span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">,</span> <span class="n">fieldType</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">));</span>
      <span class="p">}</span>
    <span class="p">}</span>
    <span class="c1">// set the output schema so downstream stages will know their input schema.</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">groupBy</span><span class="p">(</span><span class="n">StructuredRecord</span> <span class="n">input</span><span class="p">,</span> <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">groupKeyEmitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">String</span> <span class="n">val</span> <span class="o">=</span> <span class="n">input</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">val</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">return</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="k">for</span> <span class="p">(</span><span class="n">String</span> <span class="n">word</span> <span class="p">:</span> <span class="n">WHITESPACE</span><span class="p">.</span><span class="na">split</span><span class="p">(</span><span class="n">val</span><span class="p">))</span> <span class="p">{</span>
      <span class="n">groupKeyEmitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">word</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">aggregate</span><span class="p">(</span><span class="n">String</span> <span class="n">groupKey</span><span class="p">,</span> <span class="n">Iterator</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">groupValues</span><span class="p">,</span>
                        <span class="n">Emitter</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="kt">long</span> <span class="n">count</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
    <span class="k">while</span> <span class="p">(</span><span class="n">groupValues</span><span class="p">.</span><span class="na">hasNext</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">groupValues</span><span class="p">.</span><span class="na">next</span><span class="p">();</span>
      <span class="n">count</span><span class="o">++</span><span class="p">;</span>
    <span class="p">}</span>
    <span class="n">emitter</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">StructuredRecord</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">).</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;word&quot;</span><span class="p">,</span> <span class="n">groupKey</span><span class="p">).</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;count&quot;</span><span class="p">,</span> <span class="n">count</span><span class="p">).</span><span class="na">build</span><span class="p">());</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="batch-joiner-plugin">
<h2>Batch Joiner Plugin<a class="headerlink" href="#batch-joiner-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">BatchJoiner</span></code> plugin is used to join records over a batch of data.
It can be used in both batch and real-time data pipelines.
A join takes place in two steps: a <em>joinOn</em> step followed by a <em>merge</em> step.</p>
<ol class="arabic simple">
<li>In the <em>joinOn</em> step, the joiner creates a join key for each input record.
the CDAP pipeline will then take all records that have the same join key and collect them into a group.</li>
<li>The <em>merge</em> step is then called. In this step, the plugin receives a list of all the
records with same join key based on the type of join (either an <em>inner</em> or <em>outer</em> join).
It is then up to the plugin to decide what to emit, in what becomes the final output of the stage.</li>
</ol>
<p>To implement a Batch Joiner, you extend the <code class="docutils literal notranslate"><span class="pre">BatchJoiner</span></code> class. Unlike a <code class="docutils literal notranslate"><span class="pre">Transform</span></code>,
which operates on a single record at a time, a <code class="docutils literal notranslate"><span class="pre">BatchJoiner</span></code> operates on a collection of records.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to create any streams or datasets, or perform any validation
on the application configuration that is required by this plugin.</li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code>: Initialize the Batch Joiner. Guaranteed to be executed before any call
to the plugin’s <code class="docutils literal notranslate"><span class="pre">joinOn</span></code> or <code class="docutils literal notranslate"><span class="pre">merge</span></code> methods. This is called by each executor of the job.
For example, if the MapReduce engine is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">prepareRun()</span></code>: Prepares a pipeline run. This is run every time before a pipeline runs
to help set up the run. Here you can set properties such as the number of partitions
to use when joining and the join key class, if it is not known at compile time.</li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code>: Destroy any resources created by the <code class="docutils literal notranslate"><span class="pre">initialize</span></code> method. Guaranteed to be
executed after all calls to the plugin’s <code class="docutils literal notranslate"><span class="pre">joinOn</span></code> or <code class="docutils literal notranslate"><span class="pre">merge</span></code> methods have been
made. This is called by each executor of the job. For example, if the MapReduce engine
is being used, each mapper will call this method.</li>
<li><code class="docutils literal notranslate"><span class="pre">joinOn()</span></code>: This method will be called for every object that is received from the
previous stage. This method returns a join key for each object it recieves.
Objects with the same join key will be grouped together for the <code class="docutils literal notranslate"><span class="pre">merge</span></code> method.</li>
<li><code class="docutils literal notranslate"><span class="pre">getJoinConfig()</span></code>: This method will be called by the CDAP Pipeline to find out the type of join to be performed.
The config specifies which input stages are <code class="docutils literal notranslate"><span class="pre">requiredInputs</span></code>.  Records from a required input
will always be present in the <code class="docutils literal notranslate"><span class="pre">merge()</span></code> method.  Records from a non-required input will
only be present in the <code class="docutils literal notranslate"><span class="pre">merge()</span></code> method if they meet the join criteria. In other words,
if there are no required inputs, a full outer join is performed. If all inputs are
required inputs, an inner join is performed.</li>
<li><code class="docutils literal notranslate"><span class="pre">merge()</span></code>: This method is called after each object has been assigned a join key.
The method recieves a join key, an iterator over all objects with that join key, and
the stage that emitted the object. Objects emitted by this method are the output for this stage.</li>
</ul>
</div>
<div class="section" id="spark-compute-plugin">
<h2>Spark Compute Plugin<a class="headerlink" href="#spark-compute-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">SparkCompute</span></code> plugin is used to transform a collection of input records into a collection
of output records. It can be used in both batch and real-time data pipelines.
It is similar to a <code class="docutils literal notranslate"><span class="pre">Transform</span></code>, except instead of transforming its input
record by record, it transforms an entire collection. In a <code class="docutils literal notranslate"><span class="pre">SparkCompute</span></code>
plugin, you are given access to anything you would be able to do in a Spark program.</p>
<p>In order to implement a Spark Compute Plugin, you extend the <code class="docutils literal notranslate"><span class="pre">SparkCompute</span></code> class.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">transform()</span></code>: This method is given a Spark RDD (Resilient Distributed Dataset) containing
every object that is received from the previous stage. This method then performs Spark operations
on the input to transform it into an output RDD that will be sent to the next stage.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * SparkCompute plugin that counts how many times each word appears in records input to the compute stage.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">SparkCompute</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">WordCountCompute</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Counts how many times each word appears in all records input to the aggregator.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">WordCountCompute</span> <span class="kd">extends</span> <span class="n">SparkCompute</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;WordCount&quot;</span><span class="p">;</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">Schema</span> <span class="n">OUTPUT_SCHEMA</span> <span class="o">=</span> <span class="n">Schema</span><span class="p">.</span><span class="na">recordOf</span><span class="p">(</span>
    <span class="s">&quot;wordCount&quot;</span><span class="p">,</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;word&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">STRING</span><span class="p">)),</span>
    <span class="n">Schema</span><span class="p">.</span><span class="na">Field</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="s">&quot;count&quot;</span><span class="p">,</span> <span class="n">Schema</span><span class="p">.</span><span class="na">of</span><span class="p">(</span><span class="n">Schema</span><span class="p">.</span><span class="na">Type</span><span class="p">.</span><span class="na">LONG</span><span class="p">))</span>
  <span class="p">);</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The field from the input records containing the words to count.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">field</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">WordCountCompute</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// any static configuration validation should happen here.</span>
    <span class="c1">// We will check that the field is in the input schema and is of type string.</span>
    <span class="n">Schema</span> <span class="n">inputSchema</span> <span class="o">=</span> <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">getInputSchema</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">WordCount</span> <span class="n">wordCount</span> <span class="o">=</span> <span class="k">new</span> <span class="n">WordCount</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
      <span class="n">wordCount</span><span class="p">.</span><span class="na">validateSchema</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="c1">// set the output schema so downstream stages will know their input schema.</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="n">JavaRDD</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="nf">transform</span><span class="p">(</span><span class="n">SparkExecutionPluginContext</span> <span class="n">sparkExecutionPluginContext</span><span class="p">,</span>
                                             <span class="n">JavaRDD</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">javaRDD</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">WordCount</span> <span class="n">wordCount</span> <span class="o">=</span> <span class="k">new</span> <span class="n">WordCount</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
    <span class="k">return</span> <span class="n">wordCount</span><span class="p">.</span><span class="na">countWords</span><span class="p">(</span><span class="n">javaRDD</span><span class="p">)</span>
      <span class="p">.</span><span class="na">flatMap</span><span class="p">(</span><span class="k">new</span> <span class="n">FlatMapFunction</span><span class="o">&lt;</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Long</span><span class="o">&gt;</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span><span class="p">()</span> <span class="p">{</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="n">Iterable</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="nf">call</span><span class="p">(</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Long</span><span class="o">&gt;</span> <span class="n">stringLongTuple2</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
          <span class="n">List</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">output</span> <span class="o">=</span> <span class="k">new</span> <span class="n">ArrayList</span><span class="o">&lt;&gt;</span><span class="p">();</span>
          <span class="n">output</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">StructuredRecord</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="n">OUTPUT_SCHEMA</span><span class="p">)</span>
                       <span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;word&quot;</span><span class="p">,</span> <span class="n">stringLongTuple2</span><span class="p">.</span><span class="na">_1</span><span class="p">())</span>
                       <span class="p">.</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;count&quot;</span><span class="p">,</span> <span class="n">stringLongTuple2</span><span class="p">.</span><span class="na">_2</span><span class="p">())</span>
                       <span class="p">.</span><span class="na">build</span><span class="p">());</span>
          <span class="k">return</span> <span class="n">output</span><span class="p">;</span>
        <span class="p">}</span>
      <span class="p">});</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="spark-sink-plugin">
<h2>Spark Sink Plugin<a class="headerlink" href="#spark-sink-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A <code class="docutils literal notranslate"><span class="pre">SparkSink</span></code> plugin is used to perform computations on a collection of input records
and optionally write output data. It can only be used in batch data pipelines.
A <code class="docutils literal notranslate"><span class="pre">SparkSink</span></code> is similar to a <code class="docutils literal notranslate"><span class="pre">SparkCompute</span></code> plugin except that it has no output.
In a <code class="docutils literal notranslate"><span class="pre">SparkSink</span></code>, you are given access to anything you would be able to do in a Spark
program. For example, one common use case is to train a machine-learning model in this
plugin.</p>
<p>In order to implement a Spark Sink Plugin, you extend the <code class="docutils literal notranslate"><span class="pre">SparkSink</span></code> class.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">run()</span></code>: This method is given a Spark RDD (Resilient Distributed Dataset) containing every
object that is received from the previous stage. This method then performs Spark operations
on the input, and usually saves the result to a dataset.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="cm">/**</span>
<span class="cm"> * SparkSink plugin that counts how many times each word appears in records input to it and stores the result in</span>
<span class="cm"> * a KeyValueTable.</span>
<span class="cm"> */</span>
<span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">SparkSink</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="n">WordCountSink</span><span class="p">.</span><span class="na">NAME</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Counts how many times each word appears in all records input to the aggregator.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">WordCountSink</span> <span class="kd">extends</span> <span class="n">SparkSink</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;WordCount&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config properties for the plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The field from the input records containing the words to count.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">field</span><span class="p">;</span>

    <span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;The name of the KeyValueTable to write to.&quot;</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">String</span> <span class="n">tableName</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">WordCountSink</span><span class="p">(</span><span class="n">Conf</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// any static configuration validation should happen here.</span>
    <span class="c1">// We will check that the field is in the input schema and is of type string.</span>
    <span class="n">Schema</span> <span class="n">inputSchema</span> <span class="o">=</span> <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">getInputSchema</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">inputSchema</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">WordCount</span> <span class="n">wordCount</span> <span class="o">=</span> <span class="k">new</span> <span class="n">WordCount</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
      <span class="n">wordCount</span><span class="p">.</span><span class="na">validateSchema</span><span class="p">(</span><span class="n">inputSchema</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">createDataset</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">tableName</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">,</span> <span class="n">DatasetProperties</span><span class="p">.</span><span class="na">EMPTY</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">prepareRun</span><span class="p">(</span><span class="n">SparkPluginContext</span> <span class="n">sparkPluginContext</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="c1">// no-op</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">SparkExecutionPluginContext</span> <span class="n">sparkExecutionPluginContext</span><span class="p">,</span>
                  <span class="n">JavaRDD</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="n">javaRDD</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">WordCount</span> <span class="n">wordCount</span> <span class="o">=</span> <span class="k">new</span> <span class="n">WordCount</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">field</span><span class="p">);</span>
    <span class="n">JavaPairRDD</span> <span class="n">outputRDD</span> <span class="o">=</span> <span class="n">wordCount</span><span class="p">.</span><span class="na">countWords</span><span class="p">(</span><span class="n">javaRDD</span><span class="p">)</span>
      <span class="p">.</span><span class="na">mapToPair</span><span class="p">(</span><span class="k">new</span> <span class="n">PairFunction</span><span class="o">&lt;</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Long</span><span class="o">&gt;</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]&gt;</span><span class="p">()</span> <span class="p">{</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="n">Tuple2</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]&gt;</span> <span class="nf">call</span><span class="p">(</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Long</span><span class="o">&gt;</span> <span class="n">stringLongTuple2</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
          <span class="k">return</span> <span class="k">new</span> <span class="n">Tuple2</span><span class="o">&lt;&gt;</span><span class="p">(</span><span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">stringLongTuple2</span><span class="p">.</span><span class="na">_1</span><span class="p">()),</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">stringLongTuple2</span><span class="p">.</span><span class="na">_2</span><span class="p">()));</span>
        <span class="p">}</span>
      <span class="p">});</span>
    <span class="n">sparkExecutionPluginContext</span><span class="p">.</span><span class="na">saveAsDataset</span><span class="p">(</span><span class="n">outputRDD</span><span class="p">,</span> <span class="n">config</span><span class="p">.</span><span class="na">tableName</span><span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="streaming-source-plugin">
<h2>Streaming Source Plugin<a class="headerlink" href="#streaming-source-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A Streaming Source plugin is used as a source in real-time data pipelines.
It is used to fetch a Spark DStream, which is an object that represents a
collection of Spark RDDs and that produces a new RDD every batch interval of the pipeline.</p>
<p>In order to implement a Streaming Source Plugin, you extend the <code class="docutils literal notranslate"><span class="pre">StreamingSource</span></code> class.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">getStream()</span></code>: Returns the <code class="docutils literal notranslate"><span class="pre">JavaDStream</span></code> that will be used as a source in the pipeline.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">StreamingSource</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="s">&quot;Twitter&quot;</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Twitter streaming source.&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">TwitterStreamingSource</span> <span class="kd">extends</span> <span class="n">StreamingSource</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">TwitterStreamingConfig</span> <span class="n">config</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config class for TwitterStreamingSource.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">TwitterStreamingConfig</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="kd">implements</span> <span class="n">Serializable</span> <span class="p">{</span>
    <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">serialVersionUID</span> <span class="o">=</span> <span class="mi">4218063781909515444L</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">String</span> <span class="n">consumerKey</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">String</span> <span class="n">consumerSecret</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">String</span> <span class="n">accessToken</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">String</span> <span class="n">accessTokenSecret</span><span class="p">;</span>

    <span class="kd">private</span> <span class="n">String</span> <span class="n">referenceName</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">TwitterStreamingSource</span><span class="p">(</span><span class="n">TwitterStreamingConfig</span> <span class="n">config</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">config</span> <span class="o">=</span> <span class="n">config</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configurePipeline</span><span class="p">(</span><span class="n">PipelineConfigurer</span> <span class="n">pipelineConfigurer</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">pipelineConfigurer</span><span class="p">.</span><span class="na">getStageConfigurer</span><span class="p">().</span><span class="na">setOutputSchema</span><span class="p">(</span><span class="n">TwitterConstants</span><span class="p">.</span><span class="na">SCHEMA</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="n">JavaDStream</span><span class="o">&lt;</span><span class="n">StructuredRecord</span><span class="o">&gt;</span> <span class="nf">getStream</span><span class="p">(</span><span class="n">StreamingContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">JavaStreamingContext</span> <span class="n">javaStreamingContext</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getSparkStreamingContext</span><span class="p">();</span>
    <span class="c1">// lineage for this source will be tracked with this reference name</span>
    <span class="n">context</span><span class="p">.</span><span class="na">registerLineage</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">referenceName</span><span class="p">);</span>

    <span class="c1">// Create authorization from user-provided properties</span>
    <span class="n">ConfigurationBuilder</span> <span class="n">configurationBuilder</span> <span class="o">=</span> <span class="k">new</span> <span class="n">ConfigurationBuilder</span><span class="p">();</span>
    <span class="n">configurationBuilder</span><span class="p">.</span><span class="na">setDebugEnabled</span><span class="p">(</span><span class="kc">false</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setOAuthConsumerKey</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">consumerKey</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setOAuthConsumerSecret</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">consumerSecret</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setOAuthAccessToken</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">accessToken</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setOAuthAccessTokenSecret</span><span class="p">(</span><span class="n">config</span><span class="p">.</span><span class="na">accessTokenSecret</span><span class="p">);</span>
    <span class="n">Authorization</span> <span class="n">authorization</span> <span class="o">=</span> <span class="k">new</span> <span class="n">OAuthAuthorization</span><span class="p">(</span><span class="n">configurationBuilder</span><span class="p">.</span><span class="na">build</span><span class="p">());</span>

    <span class="k">return</span> <span class="n">TwitterUtils</span><span class="p">.</span><span class="na">createStream</span><span class="p">(</span><span class="n">javaStreamingContext</span><span class="p">,</span> <span class="n">authorization</span><span class="p">).</span><span class="na">map</span><span class="p">(</span>
      <span class="k">new</span> <span class="n">Function</span><span class="o">&lt;</span><span class="n">Status</span><span class="p">,</span> <span class="n">StructuredRecord</span><span class="o">&gt;</span><span class="p">()</span> <span class="p">{</span>
        <span class="kd">public</span> <span class="n">StructuredRecord</span> <span class="nf">call</span><span class="p">(</span><span class="n">Status</span> <span class="n">status</span><span class="p">)</span> <span class="p">{</span>
          <span class="k">return</span> <span class="n">convertTweet</span><span class="p">(</span><span class="n">status</span><span class="p">);</span>
        <span class="p">}</span>
      <span class="p">}</span>
    <span class="p">);</span>
  <span class="p">}</span>

  <span class="kd">private</span> <span class="n">StructuredRecord</span> <span class="nf">convertTweet</span><span class="p">(</span><span class="n">Status</span> <span class="n">tweet</span><span class="p">)</span> <span class="p">{</span>
    <span class="c1">// logic to convert a Twitter Status into a CDAP StructuredRecord</span>
  <span class="p">}</span>

<span class="p">}</span>
</pre></div>
</div>
<p class="rubric">Lineage</p>
<p>The lineage is registered using the <code class="docutils literal notranslate"><span class="pre">referenceName</span></code> provided when invoking <code class="docutils literal notranslate"><span class="pre">registerLineage()</span></code> on
<code class="docutils literal notranslate"><span class="pre">StreamingContext</span></code> in <code class="docutils literal notranslate"><span class="pre">getStream()</span></code>. Note that the <code class="docutils literal notranslate"><span class="pre">referenceName</span></code> should be a valid <code class="docutils literal notranslate"><span class="pre">DatasetId</span></code>.</p>
</div>
<div class="section" id="windower-plugin">
<h2>Windower Plugin<a class="headerlink" href="#windower-plugin" title="Permalink to this headline">🔗</a></h2>
<p>A Windower plugin is used in real-time data pipelines to create sliding windows over the data.
It does this by combining multiple micro batches into larger batches.</p>
<p>A window is defined by its <em>size</em> and its <em>slide interval</em>. Both are defined in seconds and
must be multiples of the <code class="docutils literal notranslate"><span class="pre">batchInterval</span></code> of the pipeline. The <em>size</em> defines how much data
is contained in the window. The <em>slide interval</em> defines have often a window is created.</p>
<p>For example, consider a pipeline with a <code class="docutils literal notranslate"><span class="pre">batchInterval</span></code> of 10 seconds. The pipeline uses a
<code class="docutils literal notranslate"><span class="pre">windower</span></code> that has a size of 60 and a slide interval of 30. The input into the <code class="docutils literal notranslate"><span class="pre">windower</span></code>
will be micro batches containing 10 seconds of data. Every 30 seconds, the windower will
output a batch of data containing the past 60 seconds of data, meaning the previous six
micro batches that it received as input.</p>
<p>This also means that each window output will overlap (repeat) some of the data from the
previous window. This is useful in calculating aggregates, such as how many “404” responses
did a website send out in the past ten seconds, past minute, past five minutes.</p>
<p>In order to implement a Windower Plugin, you extend the <code class="docutils literal notranslate"><span class="pre">Windower</span></code> class.</p>
<p class="rubric">Methods</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configurePipeline()</span></code>: Used to perform any validation on the application configuration
that is required by this plugin or to create any streams or datasets if the <code class="docutils literal notranslate"><span class="pre">fieldName</span></code> for a
stream or dataset is not a macro.</li>
<li><code class="docutils literal notranslate"><span class="pre">getWidth()</span></code>: Return the width in seconds of windows created by this plugin.
Must be a multiple of the <code class="docutils literal notranslate"><span class="pre">batchInterval</span></code> of the pipeline.</li>
<li><code class="docutils literal notranslate"><span class="pre">getSlideInterval()</span></code>: Get the slide interval in seconds of windows created by this plugin.
Must be a multiple of the <code class="docutils literal notranslate"><span class="pre">batchInterval</span></code> of the pipeline.</li>
</ul>
<p>Example:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="nd">@Plugin</span><span class="p">(</span><span class="n">type</span> <span class="o">=</span> <span class="n">Windower</span><span class="p">.</span><span class="na">PLUGIN_TYPE</span><span class="p">)</span>
<span class="nd">@Name</span><span class="p">(</span><span class="s">&quot;Window&quot;</span><span class="p">)</span>
<span class="nd">@Description</span><span class="p">(</span><span class="s">&quot;Creates a sliding window over the data&quot;</span><span class="p">)</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">Window</span> <span class="kd">extends</span> <span class="n">Windower</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">final</span> <span class="n">Conf</span> <span class="n">conf</span><span class="p">;</span>

  <span class="cm">/**</span>
<span class="cm">   * Config for window plugin.</span>
<span class="cm">   */</span>
  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">Conf</span> <span class="kd">extends</span> <span class="n">PluginConfig</span> <span class="p">{</span>
    <span class="kt">long</span> <span class="n">width</span><span class="p">;</span>
    <span class="kt">long</span> <span class="n">slideInterval</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="nf">Window</span><span class="p">(</span><span class="n">Conf</span> <span class="n">conf</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">conf</span> <span class="o">=</span> <span class="n">conf</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">long</span> <span class="nf">getWidth</span><span class="p">()</span> <span class="p">{</span>
    <span class="k">return</span> <span class="n">conf</span><span class="p">.</span><span class="na">width</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">long</span> <span class="nf">getSlideInterval</span><span class="p">()</span> <span class="p">{</span>
    <span class="k">return</span> <span class="n">conf</span><span class="p">.</span><span class="na">slideInterval</span><span class="p">;</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
</div>

</div>
    <div class="col-md-2">
      <div id="right-sidebar" class="bs-sidenav scrollable-y" role="complementary">
        <div id="localtoc-scrollspy">
        </div>
      </div>
    </div></div>
</div>
<!-- block main content end -->
<!-- block footer -->
<footer class="footer">
      <div class="container">
        <div class="row">
          <div class="col-md-2 footer-left"><a title="Plugin Basics" href="plugin-basics.html" />Previous</a></div>
          <div class="col-md-8 footer-center"><a class="footer-tab-link" href="../../table-of-contents/../../reference-manual/licenses/index.html">Copyright</a> &copy; 2014-2020 Cask Data, Inc.&bull; <a class="footer-tab-link" href="//docs.cask.co/cdap/6.1.1/cdap-docs-6.1.1-web.zip" rel="nofollow">Download</a> an archive or
<a class="footer-tab-link" href="//docs.cask.co/cdap">switch the version</a> of the documentation
          </div>
          <div class="col-md-2 footer-right"><a title="Plugin Presentation" href="presentation-plugins.html" />Next</a></div>
        </div>
      </div>
    </footer>
<!-- block footer end -->
<script type="text/javascript" src="../../_static/bootstrap-3.3.6/js/bootstrap.min.js"></script><script type="text/javascript" src="../../_static/js/bootstrap-sphinx.js"></script><script type="text/javascript" src="../../_static/js/abixTreeList-2.js"></script><script type="text/javascript" src="../../_static/js/cdap-dynamicscrollspy-4.js"></script><script type="text/javascript" src="../../_static/js/cdap-version-menu.js"></script><script type="text/javascript" src="../../_static/js/copy-to-clipboard.js"></script><script type="text/javascript" src="../../_static/js/jquery.mousewheel.min.js"></script><script type="text/javascript" src="../../_static/js/jquery.mCustomScrollbar.js"></script><script type="text/javascript" src="../../_static/js/js.cookie.js"></script><script type="text/javascript" src="../../_static/js/tabbed-parsed-literal-0.2.js"></script><script type="text/javascript" src="../../_static/js/cdap-onload-javascript.js"></script><script type="text/javascript" src="../../_static/js/cdap-version-menu.js"></script>
    <script src="https://cdap.gitee.io/docs/cdap/json-versions.js"/></script>
  </body>
</html>